package com.mq.mqserver.datacenter;

import com.mq.common.BinaryTool;
import com.mq.common.exception.MqException;
import com.mq.mqserver.core.MSGQueue;
import com.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/*
* 通过这个类，来针对硬盘上的消息进行管理
* data目录结构
*   meta.db
*   testQueue1  第一个文件: queue_data.txt:这里保存消息的内容
*               第二个文件: queue_stat.txt:这里保存消息的统计信息
*   testQueue2
*
*   testQueue3
* 创建各个文件路径，创建文件/队列目录，以及往文件写入，读取
* 删除队列的目录和文件
*
* */
public class MessageFileManager {
    public void init(){
        //暂时不写内容，便于后续扩展
    }
    //定义一个内部类，表示该队列的统计信息
    //优先考虑使用static，静态内部类
    static public class Stat{
        public int totalCount;
        public int validCount;
    }

    //预定消息文件所在的目录和文件名
    //这个方法是取到指定队列对应的消息文件路径
    private String getQueueDir(String queueName){
        return "./data/" + queueName;
    }
    //获取队列消息数据文件路径
     private String getQueueDataPath(String queueName){
        return getQueueDir(queueName) +"/queue_data.txt";
    }
    //获取队列消息统计的文件路径
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName) + "/queue_stat.txt";
    }
    private Stat readStat(String queueName){
        //由于当前的消息统计文件是文本文件，可以直接使用Scanner来读取文件内容
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner sc = new Scanner(inputStream);
            stat.totalCount = sc.nextInt();
            stat.validCount = sc.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    private void writeStat(String queueName,Stat stat){
        //使用PrintWirte写文件
        //OutputStream打开文件，默认情况下，会直接把原文件清空，此时相当于新的数据覆盖了旧的
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);
            printWriter.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public void createQueueFiles(String queueName) throws IOException{
        //1.创建消息队列目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            boolean ok = baseDir.mkdirs();//创建目录
            if(!ok){
                throw new IOException("创建目录失败！ baseDir" + baseDir.getAbsolutePath());
            }
        }
        //2.创建消息内容文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            boolean ok = queueDataFile.createNewFile();//创建文件
            if(!ok){
                throw new IOException("创建目录失败！ queueDataFile" + queueDataFile.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            boolean ok = queueStatFile.createNewFile();//创建文件
            if(!ok){
                throw new IOException("创建目录失败！ queueStatFile" + queueStatFile.getAbsolutePath());
            }
        }
        //4.给消息统计文件，设定初始值：0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName,stat);
    }
    //删除队列的目录和文件
    //队列也是可以被删除的，当队列删除之后，对应的消息文件自然也会删除
    public void destoryQueueFiles(String queueName) throws IOException {
        //先删除里面的文件，再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            //有任意一个删除失败，都算整体删除失败
            throw new IOException("删除队列目录和文件失败！ baseDir" + baseDir.getAbsolutePath());
        }
    }

    //检查队列目录和文件是否存在
    //比如后面生产者给broker server 生产消息了，这个消息就可能需要记录到文件上（取决于是否需要持久化）
    public boolean cheakFileExists(String queueName){
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    //这个方法用来把一个新的消息，放到队列对应的文件中
    //queue表示要把消息写入队列，message是要写的信息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查一下当前要写入的队列对应的文件是否存在
        if(!cheakFileExists(queue.getName())){
            throw new MqException("[MessageFileManager]队列文件不存在 queueName= " + queue.getName());
        }
        //2.把Message对象进行序列化，转化成二进制数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        //如果两个线程,是往同一个队列中写消息,此时需要阻塞等待.
        //如果两个线程,往不同队列中写消息,此时不需要阻塞等待.
       synchronized (queue){
           //3.先获取当前队列数据文件的长度，用来计算Message对象的offsetBeg 和 oddsetEnd
           //把新的Message 数据，写入到队列数据文件末尾，此时Message 对象的ofsetBeg就是当前文件长度+4字节
           //oddsetEnd 就是当前文件长度+4+Message长度
           File queueDataFile = new File(getQueueDataPath(queue.getName()));
           //单位是字节
           message.setOffsetBeg(queueDataFile.length()+4);
           message.setOffsetEnd(queueDataFile.length()+4+messageBinary.length);
           //4.写入消息到数据文件,注意是追加到数据文件末尾
           try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)){
               try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                   //不能直接写dataOutputStream.write(),因为dataOutputStream.write只能读入一个int
                   //使用dataOutputStream.writeInt(),能够实现逐位读取
                /*writeBuffer[0] = (byte)(v >>> 24);
                  writeBuffer[1] = (byte)(v >>> 16);
                  writeBuffer[2] = (byte)(v >>>  8);
                  writeBuffer[3] = (byte)(v >>>  0);*/
                   dataOutputStream.writeInt(messageBinary.length);
                   dataOutputStream.write(messageBinary);
               }
           }
           Stat stat = readStat(queue.getName());
           stat.validCount += 1;
           stat.totalCount += 1;
           writeStat(queue.getName(),stat);
       }
    }
    /*
     *删除消息的方法，此处删除是逻辑删除，也就是把硬盘上的数据的isvaild属性设置为0
     *思想就是：
     *1.把文件中这一段数据读出来
     *2.isvaild属性设置为0
     *3.将数据重写回文件
     */
    public void deleteMessage(MSGQueue queue, Message message) throws IOException,
            ClassNotFoundException {
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new
                    RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1. 先从⽂件中读取出 Message 的数据
                //创建盛放message容器
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() -
                        message.getOffsetBeg())];
                //移动光标到Message头部
                randomAccessFile.seek(message.getOffsetBeg());
                //读取到Message
                randomAccessFile.read(bufferSrc);
                // 2. 把当前读出来的二进制数据，转成 Message 对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3. 设置成⽆效.
                diskMessage.setIsValue((byte)0x0);
                // 4. 重新写⼊⽂件
                //Message转化为二进制
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //移动光标到Message头部，因为上述操作会读数据，光标会逐步移动到mesage末尾，
                // 所以需要重新移动光标
                randomAccessFile.seek(message.getOffsetBeg());
                //写入
                randomAccessFile.write(bufferDest);
            }
            // 更新统计⽂件
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount -= 1;
            }
            writeStat(queue.getName(), stat);
        }
    }
    //这个方法是在程序准备启动的时候，将文件中读取出所有的消息内容，加载到内存（放到一个链表中）
    //使用LinkedList,方便头删操作
    //这个方法的参数，只是一个 queueName 而不是 MSGQueue 对象，因为这个方法不需要加锁，只用 queueName 就够了
    //由于该方法是在程序启动的时候调用的，此时服务器还不能处理请求，因此不涉及多线程
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)){
                //这个变量记录当前文件光标
                long currentOffset = 0;
                //一个文件包含很多消息，所以要使用while循环读取
                while(true){
                    //1.读取当前消息的长度，这里的readInt读到末尾会抛出异常 EOFException，和之前抛出-1的流对象不一样
                    int messageSize = dataInputStream.readInt();
                    //2.按照这个长度创建容器，读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);
                    if(actualSize != messageSize){
                        //如果不匹配，说明文件格式乱了
                        throw new MqException("[MessageFileManager]队列文件不存在 queueName= " + queueName);
                    }
                    //3.将读取到的二进制数据，反序列化为 Message 对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4.判定是不是无效对象
                    if(message.getIsValue() != 0x1){
                        //虽然消息是无效数据，但是offset不要忘记更新
                        currentOffset += (4 + messageSize);
                        //无效数据直接跳过
                        continue;
                    }
                    //5.有效数据就需要把message加入链表中
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);

                }
            }catch (EOFException e){
                //这个catch是为了处理正常逻辑业务
                System.out.println("[MessageFileManager] 恢复数据完成！");
            }
        }
        return messages;
    }

    //检查当前文件是否需要针对队列的消息数据文件进行GC
    public boolean checkGC(String queueName){
        Stat stat = readStat(queueName);
        if(stat.totalCount > 2000 && (double)stat.validCount / (double) stat.totalCount < 0.5){
            return true;
        }
        return false;
    }

    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }
    //通过这个方法，真正执行消息数据文件的垃圾回收操作
    //使用复制算法完成
    //1.创建新的文件，名字就是 queue_data_new.txt
    //2.把之前消息数据文件的有效消息都读出来，写到新的文件中
    //3.删除旧的文件，在把新的文件改名回到 queue_data.txt
    //4.更新消息统计文件
    public void   GC(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        synchronized (queue){
            //由于GC比较耗时，此处统计一下GC时常，用于在系统卡顿的的时候是不是因为GC
            long gcbeg = System.currentTimeMillis();

            //1.创建一个新文件, 名字为 "queue_data_new.txt"
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常不应该存在，要是存在说明上次GC到一半，异常意外关闭了
                throw new MqException("[MessageFileManager] gc发现该队列的queue_data.txt已经存在！ queueName = " + queue.getName());
            }
            //在路径上创建新的文件夹
            boolean ok = queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败！ queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }
            //2.从旧的文件中，读取出所有的有效消息对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            //3.把有效消息写入到新的文件中
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    for(Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);// // 写入长度前缀
                        dataOutputStream.write(buffer);// 写入消息内容
                    }
                }
            }

            //4.删除旧的数据文件，并且把新的文件进行重新命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧目录失败！ queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }

            //把 queue_Data_New.txt => queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败！ queueDataNewFile = " + queueDataNewFile.getAbsolutePath()
                +queueDataNewFile.getAbsolutePath());
            }

            //5.更新统计文件
            Stat stat = new Stat();
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完毕！ queueName = " + queue.getName()
            +",time = " +(gcEnd - gcbeg) +"ms");

        }
    }


}
